package com.magicalcoder.pigeon.zookeeper;

import com.magicalcoder.pigeon.common.callback.ZookeeperNodeWatchCallback;
import com.magicalcoder.pigeon.common.dto.RmtAddress;
import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by hzhedongyu on 2017/5/31.
 * 服务端注册服务
 */
public class RegisterServer {

    private Logger logger = LoggerFactory.getLogger(RegisterServer.class);

    private ReentrantLock lock=new ReentrantLock();

    private volatile ZooKeeper zooKeeper = null;

    //绑定服务
    public void bind(String host,int port,String serviceName) throws KeeperException, InterruptedException, UnsupportedEncodingException {
        connect(new RmtAddress(host,port,serviceName));
    }
    //搜索根节点 如果没有则新增
    private void createRootNodeIfNotExist() throws KeeperException, InterruptedException {
        try {
            lock.lock();
            if(zooKeeper.exists(ZooConstant.ROOT,false)==null){
                zooKeeper.create(ZooConstant.ROOT,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }finally {
            lock.unlock();
        }
    }
    //serviceName可用快速检索出符合条件的 这里可用优化
    private void registerServer(RmtAddress rmtAddress) throws KeeperException, InterruptedException, UnsupportedEncodingException {
        String host = rmtAddress.getHost();
        int port = rmtAddress.getPort();
        String serviceName = rmtAddress.getServiceName();
        String data = String.format("%s:%d/%s",host,port,serviceName);
        zooKeeper.create(ZooConstant.SERVER_LIST,data.getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
        logger.info("server 服务注册成功");
    }


    private void connect(RmtAddress rmtAddress) {
        final CountDownLatch latch = new CountDownLatch(1);
        try {
            String server = ZookeeperProperties.getServer();
            zooKeeper = new ZooKeeper(server, 5000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    if(watchedEvent.getState() == Event.KeeperState.SyncConnected){
                        latch.countDown();
                    }else if(watchedEvent.getState() == Event.KeeperState.Expired){
                        logger.error("server 失去连接,尝试重新连接");
                        reconnect(rmtAddress);
                    }else if(watchedEvent.getState() == Event.KeeperState.Disconnected){
                        logger.error("server 失去连接,尝试重新连接");
                        reconnect(rmtAddress);
                    }else {
                        logger.info("server  zookeeper"+watchedEvent.getState());
                    }
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            latch.await();
            createRootNodeIfNotExist();
            registerServer(rmtAddress);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

        logger.info("连接成功");
    }

    private void reconnect(RmtAddress rmtAddress){
        try {
            zooKeeper.close();
            TimeUnit.SECONDS.sleep(1);
            connect(rmtAddress);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
